package com.fwmagic.spark.core.sink

import java.sql.Date
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put, Table}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.{Logger, LoggerFactory}
import scala.collection.mutable.ArrayBuffer

/**
  * 将Student的数据插入到Hbase中
  */
object StudentSinkToHbase {

    private val logger: Logger = LoggerFactory.getLogger(StudentSinkToHbase.getClass)

    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
                .setAppName(this.getClass.getSimpleName)
                //代表本地运行模式，并开启多个线程(* 表线程数)，线程数=本机核数 * 2
                .setMaster("local[*]")

        val sc = new SparkContext(conf)

        val students = new ArrayBuffer[Student]()

        //构造数据
        for (i <- 1 to 10) {
            val time: Long = System.currentTimeMillis()
            students += Student(i, "name_" + i, "pwd_" + i, new Date(time), new Date(time))
        }

        //创建RDD
        val rdd: RDD[Student] = sc.parallelize(students, 3)

        //写入数据到Hbase
        rdd.foreachPartition(fun)

        sc.stop()
    }

    private def fun(it: Iterator[Student]): Unit = {
        var con: Connection = null
        var table: Table = null
        try {
            con = getConn()
            table = con.getTable(TableName.valueOf("t_user"))
            it.foreach(stu => {
                val put = new Put(Bytes.toBytes("u_" + stu.id))
                put.addColumn(Bytes.toBytes("basic"), Bytes.toBytes("name"), Bytes.toBytes(stu.name))
                put.addColumn(Bytes.toBytes("basic"), Bytes.toBytes("password"), Bytes.toBytes(stu.password))
                table.put(put)
            })
        } catch {
            case e: Exception => {
                e.printStackTrace()
                logger.error(e.getMessage, e)
            }
        } finally {
            if (table != null) table.close()
            if (con != null) con.close()
        }
    }

    /**
      * 获取hbase链接
      *
      * @return
      */
    def getConn(): Connection = {
        val zkQuorum = "192.168.62.131:2181,192.168.62.132:2181,192.168.62.133:2181"
        val conf: Configuration = HBaseConfiguration.create()
        conf.set("hbase.zookeeper.quorum", zkQuorum)
        ConnectionFactory.createConnection(conf)
    }
}
